package com.naza.rpc.client;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.listener.Event;
import com.alibaba.nacos.api.naming.listener.EventListener;
import com.alibaba.nacos.api.naming.listener.NamingEvent;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.naza.rpc.discovery.DiscoveryListener;
import com.naza.rpc.model.ServiceInstance;
import com.naza.rpc.serialize.protostuff.ProtostuffCodecUtil;
import com.naza.rpc.serialize.protostuff.ProtostuffSerialize;
import com.naza.rpc.util.PropertiesUtil;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.CreateMode;

import java.io.ByteArrayOutputStream;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.util.ArrayList;
import java.util.List;

/**
 * zookeeper client
 *
 * @author yl
 */
public class ZookeeperClient {

    private static Object LOCK = new Object();

    private String nacosServer = PropertiesUtil.getINSTANCE().getValue("register.addr");

    private CuratorFramework client;

    private static ZookeeperClient INSTANCE = null;

    private ProtostuffCodecUtil protostuffCodecUtil = new ProtostuffCodecUtil();

    public static ZookeeperClient getINSTANCE(){
        synchronized (LOCK){
            if(null == INSTANCE){
                synchronized (LOCK){
                    if(null == INSTANCE){
                        INSTANCE = new ZookeeperClient();
                    }
                }
            }
        }
        return INSTANCE;
    }

    public ZookeeperClient(){
        init();
    }
    public void init(){
        try {
            client = CuratorClientUtils.getInstance();
        }catch (Exception e){

        }
    }

    public void registerInstance(String serviceName,ServiceInstance instance) throws NacosException {
        // 创建节点
        try {

            byte[] content = protostuffCodecUtil.encode(instance);
            client.create()
                // PERSISTENT 永久节点  EPHEMERAL 临时节点
                .withMode(CreateMode.EPHEMERAL)
                .forPath("/Naza/services/" + serviceName + "/"+instance.getIp() + ":"+instance.getPort(), content);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void deRegisterInstance(String serviceName,ServiceInstance instance) throws NacosException {
        //EPHEMERAL 临时节点
        try {
            client.delete().forPath("/Naza/services/" + serviceName + "/"+instance.getIp() + ":"+instance.getPort());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public List<ServiceInstance> getAllInstance(String serviceName) throws NacosException {
        List<ServiceInstance>  instances = new ArrayList<>();
        try {
            List<String> children = client.getChildren().forPath("/Naza/services/" + serviceName);
            for(String child : children){
                byte[] data =  client.getData().forPath("/Naza/services/" + serviceName + "/" + child);
                ServiceInstance instance = (ServiceInstance)protostuffCodecUtil.decode(data);
                instances.add(instance);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return instances;
    }

    public void monitor(String serviceName, DiscoveryListener listener) throws Exception {
        final PathChildrenCache childrenCache = new PathChildrenCache(client, "/Naza/services/" + serviceName, true);
        childrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        // 监听
        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                // 判断event type 不同事件
                if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)) {  // 添加子节点时触发,数据量不大情况下全量获取，否则考虑更新本地缓存
                    List<ServiceInstance> current = getAllInstance(serviceName);
                    listener.instanceChanged(current);
                } else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {  // 删除子节点时触发,数据量不大情况下全量获取，否则考虑更新本地缓存
                    List<ServiceInstance> current = getAllInstance(serviceName);
                    listener.instanceChanged(current);
                } else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {  // 修改子节点数据时触发,数据量不大情况下全量获取，否则考虑更新本地缓存
                    List<ServiceInstance> current = getAllInstance(serviceName);
                    listener.instanceChanged(current);
                }
            }
        });

    }
}
